package com.it.gmall.realtime.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.it.gmall.realtime.bean.DauInfo;
import com.it.gmall.realtime.util.ESUtil;
import com.it.gmall.realtime.util.KafkaUtil;
import com.it.gmall.realtime.util.OffsetManagerUtil;
import com.it.gmall.realtime.util.RedisUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.HasOffsetRanges;
import org.apache.spark.streaming.kafka010.OffsetRange;
import redis.clients.jedis.Jedis;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * desc : 日活业务。
 * 1.SparkStreaming消费kafka数据，对数据进行格式化。
 * 2.redis对当日数据进行去重。
 * 3.将每个批次新增的当日日活数据落盘到es中
 * 4.精准一次性消费：
 * 1）事务。
 * 2）手动提交偏移量+幂等性：处理的数据表多或者保存在不支持事务的数据库上。
 * 数据可能会出现数据缺失，redis存在，es中不存在：解决方式：redis存储的时候改set为zset，将redis中。
 * 幂等性：如果清空redis中的数据，那么就可能出现es数据重复，因此我们可以在插入es数据的时候指定id的方式实现幂等性。
 *
 * @author : code1997
 * @date :2021-01-2021/1/28 21:45
 */
public class DauApp {

    static OffsetRange[] offsetRanges = null;

    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("DauApp").set("spark.serializer",
                "org.apache.spark.serializer.KryoSerializer");
        JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, Durations.seconds(5));
        String topic = "gmall_start";
        String groupId = "gmall_dau";
        //获取某一个topic，某一分区的偏移量：
        Map<TopicPartition, Long> offsets = OffsetManagerUtil.getOffset(topic, groupId);
        JavaInputDStream<ConsumerRecord<String, String>> kafKaDStream = null;
        if (offsets != null && offsets.size() > 0) {
            //redis中存在offset，根据redis中进行读取
            kafKaDStream = KafkaUtil.getKafKaStream(topic, jsc, groupId, offsets);
        } else {
            //默认读取最新的
            kafKaDStream = KafkaUtil.getKafKaStream(topic, jsc, groupId);
        }
        //获取当前采集周期从kafka中消费的数据的起始偏移量以及结束偏移量值:使用原子引用设置：初始化
        JavaDStream<ConsumerRecord<String, String>> offsetDStream = kafKaDStream.transform(javaRDD -> {
            //底层封装的是kafkaRdd，注意hasOffsetRanges
            offsetRanges = ((HasOffsetRanges) javaRDD.rdd()).offsetRanges();
            System.out.println(offsetRanges[0].untilOffset() + "**************");
            return javaRDD;
        });

        //对数据进行格式化：添加事件字段
        JavaDStream<JSONObject> map = offsetDStream.map(record -> {
            String value = record.value();
            JSONObject jsonObject = JSON.parseObject(value);
            Long ts = jsonObject.getLong("ts");
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH");
            String date = format.format(new Date(ts));
            String[] split = date.split(" ");
            jsonObject.put("dt", split[0]);
            jsonObject.put("hr", split[1]);
            return jsonObject;
        });
        //利用redis对数据进行去重：以分区为单位进行过滤
        JavaDStream<JSONObject> distinctDStream = map.mapPartitions(jsonObjectIterator -> {
            Jedis jedisClient = RedisUtil.getJedisClient();
            ArrayList<JSONObject> jsonObjects = new ArrayList<>();
            while (jsonObjectIterator.hasNext()) {
                JSONObject temp = jsonObjectIterator.next();
                String dt = temp.getString("dt");
                String mid = temp.getJSONObject("common").getString("mid");
                String dauKey = "dau:" + dt;
                Long isNew = jedisClient.sadd(dauKey, mid);
                if (isNew == 1L) {
                    jedisClient.expire(dauKey, 3600 * 24);
                    jsonObjects.add(temp);
                }
            }
            jedisClient.close();
            return jsonObjects.iterator();
        });

        //批量的将数据保存到es中去:
        distinctDStream.foreachRDD(javaRDD -> {
            //以分区为单位对rdd中数据进行处理，方便批量插入
            javaRDD.foreachPartition(jsonObjectIterator -> {
                //将保存的json对象转换为样例类
                ArrayList<DauInfo> dauInfos = new ArrayList<>();
                while (jsonObjectIterator.hasNext()) {
                    JSONObject next = jsonObjectIterator.next();
                    JSONObject common = next.getJSONObject("common");
                    DauInfo dauInfo = new DauInfo(common.getString("mid"),
                            common.getString("uid"),
                            common.getString("ar"),
                            common.getString("ch"),
                            common.getString("vc"),
                            next.getString("dt"),
                            next.getString("hr"),
                            "00",
                            next.getLong("ts"));
                    dauInfos.add(dauInfo);
                }
                String dt = new SimpleDateFormat("yyyy-MM-dd").format(new Date());
                ESUtil.bulkInsert(dauInfos, "gmall_dau_info_" + dt);

            });
            //提交偏移量到offset中去
            OffsetManagerUtil.setOffset(topic, groupId, offsetRanges);
        });


        try {
            jsc.start();
            jsc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
